package com.shujia.profile

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Table}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}


trait UserProfile {

  //标签名
  var flagName: String = _

  //日期参数
  var ds: String = _

  /**
   * 用于提取用户画像的接口
   *
   * @param spark ： spark环境
   * @return
   */
  def extractProfile(spark: SparkSession): DataFrame

  /**
   * 执行生成用户用户画像的方法
   *
   * @param spark ：spark环境
   */
  def run(spark: SparkSession, _ds: String): Unit = {

    //给分区时间参数赋值
    ds = _ds

    //调用子类的方法，生成用户画像
    val df: DataFrame = extractProfile(spark)
    //将用户画像保存到hbase中
    profileSaveHbase(df, flagName)
  }


  /**
   * 通用的将数据保存到hbase中的方法
   *
   * @param df       ：需要保存的DF
   * @param flagName : 标签名
   * @tparam T ：标签类型
   */
  def profileSaveHbase(df: DataFrame, flagName: String): Unit = {
    df.foreachPartition(iter => {
      //创建hbase连接
      // 创建连接需要给定连接信息，所以需要给定配置类
      val conf: Configuration = new Configuration

      // 由于对HBASE进行操作时，需要连接其Zookeeper 那么需要传入Zookeeper的IP
      conf.set("hbase.zookeeper.quorum", "node1,node2,master")

      // 获取连接对象 通过ConnectionFactory来创建连接对象
      val connection: Connection = ConnectionFactory.createConnection(conf)

      //获取表
      val table: Table = connection.getTable(TableName.valueOf("userprofile"))

      iter.foreach((row: Row) => {
        //取出用户编号
        val id: Long = row.getAs[Long]("id")
        //获取标签值
        val flagValue: Any = row.getAs[Any](flagName)

        val put = new Put(id.toString.getBytes())

        //将标签值转换成String类型
        put.addColumn("info".getBytes(), flagName.getBytes(), flagValue.toString.getBytes())

        table.put(put)
      })

      connection.close()
    })
  }
}
